#include "includes.h"
#include "structs.h"
#include "declares.h"


int grape_init( struct grape_plugin_t *plugin )
{
        grape_errexit_if(!plugin, 4, "Not plugin any algorithm into grape, quit\n");
        grape_plugin=plugin;

        // pthread attributes
        pthread_attr_init(&grape_var.joinable_attr);
        pthread_attr_setdetachstate(&grape_var.joinable_attr, PTHREAD_CREATE_JOINABLE);

        // timer 
        igraph_vector_ptr_init(&grape_graph.all_queries, 0);
        init_timers();
        igraph_vector_ptr_init(&grape_var.timer,0);
        igraph_vector_ptr_set_item_destructor(&grape_var.timer, free);


        for(int i=0;i<grape_data.nfrag;++i) {
                grape_var.args[i].global_adjlist=&grape_graph.global_adjlist;
                grape_var.args[i].fragment=&grape_var.fragments[i];
                grape_var.args[i].messages=&grape_var.messages[i];
                grape_var.args[i].nfrag=grape_data.nfrag;
                grape_var.args[i].vertices=&grape_graph.vertices;
                grape_var.args[i].edges=&grape_graph.edges;
                msg_buf_init(&grape_var.messages[i], grape_data.nfrag);
        } 
        return 0;
}

/* is_finish:
 * Check if all the work is done, signaled by no new IN msg
 * is generated by sync_message OR reaching max_step
 * Input: argumets array and its len, transfering IN msg 
 *      and step limit argumets
 * Ouput: bool, whether the work is done
 */

int is_finish(struct args_t *args, int nfrag)
{
        int halt_cnt=0;

        for(int i=0;i<nfrag;++i) {
                int size=get_in_msg_size(args[i].messages);
                if(0 == size) 
                        halt_cnt++;
        }
        return (nfrag == halt_cnt);
}

/* run_on_fragments:
 * run the algorithm specified by peval, inceval and assemble
 * Input: 
 *      peval: PARTIAL EVALUATION
 *      inceval: INCREMENTAL EIVALUATION
 *      args: argumets array that the algorithm 
 *      will be run on each fragment 
 *      params: other parameters that provided input
 *      or tuning to the function
 */

void grape_summary()
{
        zlog_info(grape_proc_logger,"-----------------------------GRAPE-SUMMARY------------------------------");
        zlog_info(grape_proc_logger, "algorithm = %s"     ,grape_app.app_name);
        zlog_info(grape_proc_logger, "vfile = %s"     ,arguments.vfile);
        zlog_info(grape_proc_logger,"efile = %s"      ,arguments.efile);
        zlog_info(grape_proc_logger,"rfile = %s"    ,arguments.rfile);
        zlog_info(grape_proc_logger,"query = %s" ,arguments.query_file);

        zlog_info(grape_proc_logger,"nfrag = %d"    ,arguments.nfrag);
        zlog_info(grape_proc_logger,"ncon  = %d"    ,arguments.ncon);
        zlog_info(grape_proc_logger,"output-file = %s"   ,arguments.output_file);
        zlog_info(grape_proc_logger,"graph loading time: %lf s", get_timer(ALL_LOAD_TIMER));
        zlog_info(grape_proc_logger,"graph evaluation time: %lf s(excluding graph loading time)", get_timer(ALL_EVAL_TIMER));
        zlog_info(grape_proc_logger,"------------------------------------------------------------------------");


}

void grape_step_summary()
{

        for(int i=0;i<grape_data.nstep;++i) {
                zlog_info(grape_proc_logger,"super_step %d time record:",i);
                struct step_timer_t *st=VECTOR(grape_var.timer)[i];
                zlog_info(grape_proc_logger,"sync_message time: %lf s", st->synct);
                zlog_info(grape_proc_logger,"evaluation time: %lf s", st->evalt);
                zlog_info(grape_proc_logger,"------------------------------------------------------------------------");
        }
        reset_step_timer(&grape_var.timer);
}

int run_on_fragments()

{
        assert(grape_plugin);
        reset_timer(ALL_LOAD_TIMER);        

        /* vfile2vertices */
        grape_load_vertices();

        /* efile2edges */
        grape_load_edges();

        /* rfile2fragtable */        
        grape_load_fragtable();


        // build global adjlist 
        grape_build_global_adjlist();

        /* load_fragment */
        grape_load_fragments();

        /* load queries */
        grape_load_queries();
        stop_timer(ALL_LOAD_TIMER);

        // further init  
        for(int i=0;i<grape_data.nfrag;++i) {
                grape_plugin->result_init_func(&grape_var.args[i],&grape_var.results[i]);
                grape_var.args[i].result=grape_var.results[i];
        }  

        zlog_info(grape_proc_logger,"evaluation begin...");
        reset_timer(ALL_EVAL_TIMER);

        for(int i=0;i<grape_data.qn;++i) {

                reset_timer(EACH_Q_TIMER);
                zlog_info(grape_proc_logger, "processing query %d...", i+1);

                // broadcast query
                zlog_info(grape_proc_logger, "broadcasting query %d...", i+1);
                reset_timer(BCAST_TIMER);
                for(int j=0;j<grape_data.nfrag;++j) {
                        void *p=VECTOR(grape_graph.all_queries)[i];
                        grape_var.args[j].query=grape_plugin->query_copy_func(p);
                }
                stop_timer(BCAST_TIMER);
                zlog_info(grape_proc_logger, "query %d is broadcasted in %lf s", i+1, get_timer(BCAST_TIMER));

                // main loop
                for(int super_step=0;super_step<grape_data.max_step;++super_step)
                {

                        int rs=0;
                        void *status=NULL;

                        if(super_step==0) {
                                zlog_info(grape_proc_logger, "[coordinator]: PEval begins");
                        }
                        else{
                                zlog_info(grape_proc_logger, "[coordinator]: IncEval %d begins", super_step);
                        }
                        // create threads to do the tasks
                        reset_timer(EVAL_TIMER);

                        // parallel version
                        if(grape_data.parallel) {
                                for(int i=0;i<grape_data.nfrag;++i) {
                                        if(super_step==0) {
                                                // PEval
                                                rs = pthread_create(&grape_var.threads[i], &grape_var.joinable_attr,
                                                                grape_plugin->peval, &grape_var.args[i]);
                                        }
                                        else {
                                                // IncEval
                                                rs = pthread_create(&grape_var.threads[i], &grape_var.joinable_attr,
                                                                grape_plugin->inceval, &grape_var.args[i]);
                                        }
                                        grape_errexit_if(rs, 3, "pthread_create failed\n");
                                }

                                // join all the threads for sync_message
                                for(int i=0;i<grape_data.nfrag;++i) {
                                        rs = pthread_join(grape_var.threads[i], &status);
                                        grape_errexit_if(rs, 3, "pthread_join failed\n");
                                }
                        }
                        // serial debug version
                        else {
                                for(int i=0;i<grape_data.nfrag;++i) {
                                        if(super_step==0) {
                                                // PEval
                                                grape_plugin->peval(&grape_var.args[i]);
                                        }
                                        else {
                                                // IncEval
                                                grape_plugin->inceval(&grape_var.args[i]);
                                        }
                                }

                        }
                        stop_timer(EVAL_TIMER);

                        if(super_step==0) {
                                zlog_info(grape_proc_logger,"[coordinator]: PEval finishes in %lf s",get_timer(EVAL_TIMER));
                        }
                        else {
                                zlog_info(grape_proc_logger, "[coordinator]: IncEval %d finishes in %lf s", super_step,
                                                get_timer(EVAL_TIMER));
                        }

                        printl_msg_buf_all(grape_var.messages);

                        zlog_info(grape_proc_logger , "[coordinator]: sync_message begins");
                        reset_timer(SYNC_TIMER);
                        sync_message(grape_var.args, grape_data.nfrag);
                        stop_timer(SYNC_TIMER);
                        zlog_info(grape_proc_logger, "[coordinator]: sync_message finishes in %lf s", get_timer(SYNC_TIMER));

                        struct step_timer_t *st = make_timer(get_timer(SYNC_TIMER), get_timer(EVAL_TIMER));
                        add_step_time(&grape_var.timer, st);

                        // check for termination
                        if(is_finish(grape_var.args, grape_data.nfrag)) {
                                grape_data.nstep=igraph_vector_ptr_size(&grape_var.timer);
                                grape_step_summary();
                                break;
                        }

                }
                char *dat=va_format("%s.%s.query_%d.dat", arguments.output_file, grape_app.app_name, i+1);
                grape_file.dat=fopen(dat, "w");
                grape_errexit_if(!grape_file.dat, 1, GRAPE_EFILE, "output file", dat, "writing");

                // assemble
                zlog_info(grape_proc_logger, "[coordinator]: assemble begins");
                reset_timer(ASSEM_TIMER);
                grape_plugin->assemble(grape_var.args, grape_data.nfrag);
                stop_timer(ASSEM_TIMER);
                zlog_info(grape_proc_logger, "[coordinator]: assemble finishes in %lf s", get_timer(ASSEM_TIMER));

                stop_timer(EACH_Q_TIMER);
                zlog_info(grape_proc_logger, "query %d is processed in %lf s, result is written to %s", i+1,get_timer(EACH_Q_TIMER), dat);
                zlog_info(grape_proc_logger,"------------------------------------------------------------------------");
                free(dat);
                fclose(grape_file.dat);
        }

        zlog_info(grape_proc_logger,"evaluation finishes");
        stop_timer(ALL_EVAL_TIMER);

        grape_summary();
        /* return 0; */
        pthread_exit(NULL);
}
int grape_fini()
{
        // shutdown zlog
        zlog_fini();

        // close all files 
        fclose(grape_file.vfile);
        fclose(grape_file.efile);
        fclose(grape_file.query_file);
        fclose(grape_file.rfile);

        // destroy global var
        igraph_vector_ptr_destroy_all(&grape_graph.all_queries);
        free(grape_app.app_name);
        igraph_vector_ptr_destroy(&grape_graph.edges);
        igraph_vector_ptr_destroy(&grape_graph.vertices);
        igraph_vector_int_destroy(&grape_graph.fragtable); 
        for(int i=0;i<grape_data.nfrag;++i) {
                fragment_free(&grape_var.fragments[i]);
                msg_buf_destroy(&grape_var.messages[i]);
                grape_plugin->result_destroy_func(grape_var.results[i]);
        }
        return 0;
}

void grape_load_vertices()
{

        /* vfile2vertices */
        zlog_info(grape_proc_logger,"loading vfile...");
        reset_timer(LOAD_TIMER);
        int vn=vfile2vertices(grape_file.vfile,grape_data.ncon,&grape_graph.vertices);
        stop_timer(LOAD_TIMER);
        zlog_info(grape_proc_logger,"vfile is loaded in %lf s",get_timer(LOAD_TIMER));
        grape_data.vn = vn;
}

void grape_load_fragments()
{

        zlog_info(grape_proc_logger,"loading fragment...");
        reset_timer(LOAD_TIMER);
        load_fragment( 
                        grape_data.nfrag,
                        &grape_graph.vertices, 
                        &grape_graph.edges, 
                        &grape_graph.fragtable, 
                        &grape_graph.global_adjlist, 
                        grape_var.fragments);
        stop_timer(LOAD_TIMER);
        zlog_info(grape_proc_logger,"fragment is loaded in %lf s", get_timer(LOAD_TIMER));

}

void grape_load_fragtable()
{
        zlog_info(grape_proc_logger,"loaded rfile...");
        reset_timer(LOAD_TIMER);
        rfile2fragtable(grape_file.rfile,grape_data.vn, &grape_graph.fragtable);
        stop_timer(LOAD_TIMER);
        zlog_info(grape_proc_logger,"rfile is loaded in %lf s", get_timer(LOAD_TIMER));
}
void grape_load_edges()
{
        /* efile2edges */
        zlog_info(grape_proc_logger,"loading efile...");
        reset_timer(LOAD_TIMER);
        int en=efile2edges(grape_file.efile, &grape_graph.edges); 
        stop_timer(LOAD_TIMER);
        zlog_info(grape_proc_logger,"efile is loaded in %lf s", get_timer(LOAD_TIMER));
        grape_data.en=en;
}

void grape_build_global_adjlist()
{

        // build global adjlist 
        vector2my_adjlist(&grape_graph.vertices,
                        &grape_graph.edges,
                        &grape_graph.global_adjlist);

}

void grape_load_queries()
{

        /* load queries */
        zlog_info(grape_proc_logger, "loading queries...");
        reset_timer(LOAD_TIMER);
        grape_plugin->query_load_func(grape_file.query_file, &grape_graph.all_queries);
        stop_timer(LOAD_TIMER);
        zlog_info(grape_proc_logger, "queries is loaded in %lf s", get_timer(LOAD_TIMER));
        grape_data.qn=igraph_vector_ptr_size(&grape_graph.all_queries);
}

